/*
 * Title:        CloudScheduler Toolkit
 * Description:  a Toolkit for Modeling and Simulation of Job Scheduling and Resource Provisioning in Cloud System
 * Licence:      GPL - http://www.gnu.org/copyleft/gpl.html
 * Author:       Fuhui Wu
 *
 * Copyright (c) 2013-2014, The National University of Defense Technology, China
 */

package org.nudt.jCloud.example.resourceStealing2;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.nudt.jCloud.infrastructure.Cluster;
import org.nudt.jCloud.infrastructure.Slot;
import org.nudt.jCloud.infrastructure.SlotType;
import org.nudt.jCloud.resourceManager.ReservationRM;
import org.nudt.jCloud.simulation.Batch;
import org.nudt.jCloud.simulation.CloudSchedSIM;
import org.nudt.jCloud.simulation.Framework;
import org.nudt.jCloud.simulation.SIMTag;
import org.nudt.jCloud.simulation.ScaleData;
import org.nudt.jCloud.simulation.SimEvent;
import org.nudt.jCloud.simulation.UserBroker;
import org.nudt.jCloud.workload.Task;

public class StealingRM extends ReservationRM{
	private Map<Framework, Double> reservationRatio;
	private int totalSlotNum;
	private boolean initialized;

	public StealingRM(String name, Cluster cluster) {
		super(name, cluster);
		reservationRatio = new HashMap<Framework, Double>();
		initialized = false;
	}
	
	public void initSlotRatio(){
		double avgRatio = ((double)1.0)/getFrwkList().size();
		for(Framework frwk : getFrwkList()){
			getReservationRatio().put(frwk, avgRatio);
		}
		setInitialized(true);
	}
	
	public void resetTotalSlotNum(){
		int totalSlotNum = 0;
		for(Framework frwk : getFrwkList()){
			totalSlotNum += frwk.busySlotNum()+frwk.freeSlotNum();
		}
		setTotalSlotNum(totalSlotNum);
	}
						
	public Framework mostOcpWithFreeNoReq(List<Framework> requestFrwkList,
			List<Framework> registeredFrwkList) {
		Framework frwk = null;
		double slotRatio = 0;
		for(Framework curFrwk : registeredFrwkList){
			if(requestFrwkList.contains(curFrwk) == false){
				int curFreeNum = curFrwk.freeSlotNum();
				if(curFreeNum > 0){
					double curSlotNum = curFrwk.busySlotNum()+curFreeNum;
					double curSlotRatio = (curSlotNum/totalSlotNum)/reservationRatio.get(curFrwk);
					if(curSlotRatio>slotRatio){
						frwk = curFrwk;
						slotRatio = curSlotRatio;
					}
				}
			}
		}
		return frwk;
	}
	
	public Framework mostOcpwithFreeInReq(List<Framework> requestFrwkList){
		Framework frwk = null;
		double slotRatio = 0;
		for(Framework curFrwk : requestFrwkList){
			int curFreeNum = curFrwk.freeSlotNum();
			if(curFreeNum > 0){
				double curSlotNum = curFrwk.busySlotNum()+curFreeNum;
				double curSlotRatio = (curSlotNum/totalSlotNum)/reservationRatio.get(curFrwk);
				if(curSlotRatio>slotRatio){
					frwk = curFrwk;
					slotRatio = curSlotRatio;
				}
			}
		}
		return frwk;
	}
	
	public Framework leastOcpUnSatisfied(List<ScaleData> scaleDataList){
		Framework frwk = null;
		double slotRatio = Double.MAX_VALUE;
		for(ScaleData scaleData : scaleDataList){
			UserBroker userBroker = (UserBroker)CloudSchedSIM.getEntityById(scaleData.getUserBrokerId());
			Framework curFrwk = userBroker.getFrwk();
			for(SlotType slotType : scaleData.getToLeaseSlot().keySet()){
				int freeNum = curFrwk.freeSlotNum();
				if(scaleData.getToLeaseSlot().get(slotType)>freeNum){
					double curSlotNum = curFrwk.busySlotNum()+curFrwk.freeSlotNum();
					double curSlotRatio = (curSlotNum/totalSlotNum)/reservationRatio.get(curFrwk);
					if(curSlotRatio<slotRatio){
						frwk = curFrwk;
						slotRatio = curSlotRatio;
					}
				}
			}
		}
		return frwk;
	}
	
	public Framework leastOcpUnSatisfiedBelowThresh(List<ScaleData> scaleDataList){
		Framework frwk = null;
		double slotRatio = Double.MAX_VALUE;
		for(ScaleData scaleData : scaleDataList){
			UserBroker userBroker = (UserBroker)CloudSchedSIM.getEntityById(scaleData.getUserBrokerId());
			Framework curFrwk = userBroker.getFrwk();
			for(SlotType slotType : scaleData.getToLeaseSlot().keySet()){
				int freeNum = curFrwk.freeSlotNum();
				double curSlotNum = freeNum + curFrwk.busySlotNum();
				double curSlotRatio = (curSlotNum/totalSlotNum)/reservationRatio.get(curFrwk);
				if(scaleData.getToLeaseSlot().get(slotType)>freeNum && curSlotRatio<1){
					if(curSlotRatio<slotRatio){
						frwk = curFrwk;
						slotRatio = curSlotRatio;
					}
				}
			}
		}
		return frwk;
	}
	
	public Framework mostOcpSatisfied(Framework stealingFrwk, List<ScaleData> scaleDataList){
		Framework frwk = null;
		double slotRatio = 0;
		for(ScaleData scaleData : scaleDataList){
			UserBroker userBroker = (UserBroker)CloudSchedSIM.getEntityById(scaleData.getUserBrokerId());
			Framework curFrwk = userBroker.getFrwk();
			for(SlotType slotType : scaleData.getToLeaseSlot().keySet()){
				int toLeaseNum = scaleData.getToLeaseSlot().get(slotType);
				if(stealingFrwk != curFrwk){
					if(curFrwk.freeSlotNum()>toLeaseNum){
						double curSlotNum = curFrwk.busySlotNum()+curFrwk.freeSlotNum();
						double curSlotRatio = (curSlotNum/totalSlotNum)/reservationRatio.get(curFrwk);
						if(slotRatio < curSlotRatio){
							frwk = curFrwk;
							slotRatio = curSlotRatio;
						}
					}
				}
			}
		}
		return frwk;
	}
	
	public Framework mostOcpOverThresh(List<ScaleData> scaleDataList){
		Framework frwk = null;
		double slotRatio = 0;
		for(ScaleData scaleData : scaleDataList){
			UserBroker userBroker = (UserBroker)CloudSchedSIM.getEntityById(scaleData.getUserBrokerId());
			Framework curFrwk = userBroker.getFrwk();
			double curSlotNum = curFrwk.busySlotNum()+curFrwk.freeSlotNum();
			double curSlotRatio = (curSlotNum/totalSlotNum)/reservationRatio.get(curFrwk);
			if(curSlotRatio>1 && curFrwk.freeSlotNum()>0){
				if(slotRatio<curSlotRatio){
					frwk = curFrwk;
					slotRatio = curSlotRatio;
				}
			}
		}
		return frwk;
	}
	
	public Framework mostOcpOverThreswithoutFree(Framework preemptingFrwk,
			List<Framework> registeredFrwkList, Map<Framework, List<Slot>>  preemptSlotSet){
		Framework frwk = null;
		double slotRatio = 0;
		for(Framework curFrwk : registeredFrwkList){
			if(preemptingFrwk == curFrwk){
				continue;
			}else{
				int preemptNum = 0;
				if(preemptSlotSet.containsKey(curFrwk)){
					preemptNum += preemptSlotSet.get(curFrwk).size();
				}
				int busyNum = curFrwk.busySlotNum();
				double curSlotNum = busyNum-preemptNum;
				double curSlotRatio = (curSlotNum/totalSlotNum)/reservationRatio.get(curFrwk);
				if(slotRatio <curSlotRatio && curSlotRatio>1){
					frwk = curFrwk;
					slotRatio = curSlotRatio;
				}
			}
		}
		return frwk;
	}
	
	public ScaleData leastOcpInReq(List<ScaleData> scaleDataList){
		ScaleData scaleData = null;
		double slotRatio = Double.MAX_VALUE;
		for(ScaleData curScaleData : scaleDataList){
			UserBroker userBroker = (UserBroker)CloudSchedSIM.getEntityById(curScaleData.getUserBrokerId());
			Framework curFrwk = userBroker.getFrwk();
			double curSlotNum = curFrwk.busySlotNum()+curFrwk.freeSlotNum();
			double curSlotRatio = (curSlotNum/totalSlotNum)/reservationRatio.get(curFrwk);
			if(curSlotRatio<slotRatio){
				scaleData = curScaleData;
				slotRatio = curSlotRatio;
			}
		}
		return scaleData;
	}
	
	public ScaleData scaleDataInLeasing(List<ScaleData> scaleDataList, Framework frwk){
		for(ScaleData scaleData : scaleDataList){
			if(scaleData.getUserBrokerId()==frwk.getUserBroker().getId()){
				return scaleData;
			}
		}
		return null;
	}
	
	public void stealOneSlot(Framework stealingFrwk, Framework stolenFrwk){
		Slot slot = null;
		for(SlotType slotType : stolenFrwk.getFreeSlots().keySet()){
			slot = stolenFrwk.getFreeSlots().get(slotType).get(0);
			stolenFrwk.getFreeSlots().get(slotType).remove(slot);
			if(stolenFrwk.getFreeSlots().get(slotType).size()<=0){
				stolenFrwk.getFreeSlots().remove(slotType);
			}
			break;
		}
		stealingFrwk.leaseSlot(slot);
	}
	
	public void lease(List<ScaleData> scaleDataList){
		while(true){
			Framework leasingFrwk = leastOcpUnSatisfied(scaleDataList);
			boolean noUnLeasedRes = false;
			if(leasingFrwk == null){
				break;
			}else{
				ScaleData scaleData = scaleDataInLeasing(scaleDataList, leasingFrwk);
				for(SlotType slotType : scaleData.getToLeaseSlot().keySet()){
					List<Slot> leasedSlotList = provideSlots(slotType, 1);
					if(leasedSlotList.size()>0){
						leasingFrwk.leaseSlot(slotType, leasedSlotList);
					}else{
						noUnLeasedRes = true;
					}
				}
				if(noUnLeasedRes == true){
					break;
				}
			}
		}
	}
	
	public List<Framework> requestFrwk(List<ScaleData> scaleDataList){
		List<Framework > requestFrwkList = new ArrayList<Framework>();
		for(ScaleData scaleData : scaleDataList){
			UserBroker userBroker = (UserBroker)CloudSchedSIM.getEntityById(scaleData.getUserBrokerId());
			requestFrwkList.add(userBroker.getFrwk());
		}
		return requestFrwkList;
	}
	
	public void stealFromNoReq(List<Framework> requestFrwkList, List<ScaleData> scaleDataList){
		while(true){
			Framework stolenFrwk = mostOcpWithFreeNoReq(requestFrwkList, getFrwkList());
			if(stolenFrwk == null){
				break;
			}else{
				Framework stealingFrwk = leastOcpUnSatisfied(scaleDataList);
				if(stealingFrwk == null){
					break;
				}else{
					stealOneSlot(stealingFrwk, stolenFrwk);
				}
			}
		}
	}
	
	public void stealFromReqSatisfied(List<ScaleData> scaleDataList){
		while(true){
			Framework stealingFrwk = leastOcpUnSatisfied(scaleDataList);
			if(stealingFrwk == null){
				break;
			}else{
				Framework stolenFrwk = mostOcpSatisfied(stealingFrwk, scaleDataList);
				if(stolenFrwk == null){
					break;
				}else{
					stealOneSlot(stealingFrwk, stolenFrwk);
				}
			}
		}
	}
	
	public void stealFromReqOverThresh(List<ScaleData> scaleDataList){
		while(true){
			Framework stealingFrwk = leastOcpUnSatisfiedBelowThresh(scaleDataList);
			if(stealingFrwk == null){
				break;
			}else{
				Framework stolenFrwk = mostOcpOverThresh(scaleDataList);
				if(stolenFrwk == null){
					break;
				}else{
					stealOneSlot(stealingFrwk, stolenFrwk);
				}
			}
		}
	}
	
	public void preempt(List<Framework> requestFrwkList, List<ScaleData> scaleDataList){
		Map<Framework, List<Slot>> preemptSlotSet = new HashMap<Framework, List<Slot>>();
		while(true){
			Framework preemptingFrwk = leastOcpUnSatisfiedBelowThresh(scaleDataList);
			if(preemptingFrwk == null){
				break;
			}else{
				Framework preemptedFrwk = mostOcpOverThreswithoutFree(preemptingFrwk, getFrwkList(), preemptSlotSet);
				if(preemptedFrwk == null){
					break;
				}else{
					Collections.sort(preemptedFrwk.getJobScheduler().getScheduledBatches(), new Batch.SortBatchByScheduleTime());
					Slot slot = null;
					for(Batch batch : preemptedFrwk.getJobScheduler().getScheduledBatches()){
						for(Task task : batch.getTaskList()){
							Slot curSlot = task.getSlot();
							if(preemptSlotSet.containsKey(preemptedFrwk)){
								if(preemptSlotSet.get(preemptedFrwk).contains(curSlot)==false){	
									slot = curSlot;
									break;
								}
							}else{
								slot = curSlot;
								break;
							}
						}
						if(slot!=null){
							break;
						}
					}
					preemptingFrwk.leaseSlot(slot);
					if(preemptSlotSet.containsKey(preemptedFrwk)){
						preemptSlotSet.get(preemptedFrwk).add(slot);
					}else{
						preemptSlotSet.put(preemptedFrwk, new ArrayList<Slot>());
						preemptSlotSet.get(preemptedFrwk).add(slot);
					}
				}				
			}
		}
		
		for(Framework frwk : preemptSlotSet.keySet()){
			PreemptionJobScheduler jobScheduler = (PreemptionJobScheduler) frwk.getJobScheduler();
			jobScheduler.preemptSlot(preemptSlotSet.get(frwk));
			if(requestFrwkList.contains(frwk)==false){
				UserBroker userBroker = frwk.getUserBroker();
				List<SimEvent> events = CloudSchedSIM.getFuture().get(userBroker.getId(), SIMTag.UPDATE_OR_SUBMIT_JOB);
				for(int j=0; j<events.size(); ++j){
					CloudSchedSIM.getFuture().remove(events.get(j));
				}
				jobScheduler.nextEvent();		
			}
		}			
	}
	
	public Map<Framework, Double> getReservationRatio() {
		return reservationRatio;
	}

	public void setReservationRatio(Map<Framework, Double> reservationRatio) {
		this.reservationRatio = reservationRatio;
	}

	public int getTotalSlotNum() {
		return totalSlotNum;
	}

	public void setTotalSlotNum(int totalSlotNum) {
		this.totalSlotNum = totalSlotNum;
	}

	@Override
	public void scaleResource(List<ScaleData> scaleDataList) {
		// TODO Auto-generated method stub
		
	}

	public boolean isInitialized() {
		return initialized;
	}

	public void setInitialized(boolean initialized) {
		this.initialized = initialized;
	}
}
